package com.googlecode.jlogalizer.mapred.log;

import java.io.File;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.lib.NLineInputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.googlecode.jlogalizer.mapred.log.format.ClientIdMultipleTextOutputFormat;
import com.googlecode.jlogalizer.mapred.log.map.LogMapper;
import com.googlecode.jlogalizer.mapred.log.reduce.LogReducer;
import com.googlecode.jlogalizer.mapred.util.MapReduceFileUtils;


public class LogMapReduceJob extends Configured implements Tool{

	private static void initJob(String jobName, Configuration config, String inputPath, String outputPath) throws IOException, InterruptedException, ClassNotFoundException{
	
		MapReduceFileUtils.deleteDirectory(new File(outputPath));
		/*Job job=new Job(config, jobName);
		job.setJarByClass(LogMapReduceJob.class);
		job.setMapperClass(LogMapper.class);
		job.setReducerClass(LogReducer.class);
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		FileInputFormat.addInputPath(job, new Path(inputPath));
		FileOutputFormat.setOutputPath(job, new Path(outputPath));
		
		job.waitForCompletion(true);*/
		
		JobConf job = new JobConf(config, LogMapReduceJob.class);
				     
		job.setJobName(jobName);
				     
		FileInputFormat.addInputPath(job, new Path(inputPath));
		FileOutputFormat.setOutputPath(job, new Path(outputPath));
		 
		job.setMapperClass(LogMapper.class);
		job.setReducerClass(LogReducer.class);
		
		//job.setOutputKeyClass(Text.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		 
		job.setInputFormat(NLineInputFormat.class);
		//job.setOutputFormat(TextOutputFormat.class);
		job.setOutputFormat(ClientIdMultipleTextOutputFormat.class);
		 
		JobClient.runJob(job);
		
	}
	
	public static void main(String[] args) throws Exception {
		ToolRunner.run(new LogMapReduceJob(), args);
	}

	@Override
	public int run(String[] args) throws Exception {
		
		Configuration config= getConf();
		System.out.println("Args [0] :"+args[0]);
		System.out.println("Args [1] :"+args[1]);
		System.out.println("Args [2] :"+args[2]);
		config.set("mapred.line.input.format.linespermap", "00");
		config.set("JOB_NAME", args[0]);
		/*config.set("log_file_type", "apache_access_log");
		config.setInt("position_in_log_file", 1);*/
		
		initJob(args[0], config, args[1], args[2]);
		return 0;
	}
	
	
}
